package com.atguigu.flink.streamapi.trans;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/1/29
 *
 *    OperatorChain: 算子链
 *        自动执行。
 *
 *        算子链节省slot! 节省资源。
 *        减少数据传输的网络消耗(时间消耗，IO资源消耗)
 *   ---------------------------
 *      合并的必要条件。
 *         条件1: 上下游必须是forward传输。
 *                  上下游的parilism必须是一致的。
 *                  必须是1:1传输
 *                      不能是1:N   shuffle,keyBy,reblance,rescale,broadcast
 *                      不能是N:1   global
 *
 *         条件2:  要合并的算子必须在一个 SlotSharingGroup(槽位共享组)
 *                      SlotSharingGroup： 就是一个标记，用来标记算子是否合得来，是否可以配对到一起！
 *                      默认情况，所有的算子的SlotSharingGroup都是一致的，都是可以合并的。
 *                      SlotSharingGroup有向下游传递的属性。
 *
 *                 很少用。
 *
 *  -------------------------------
 *      断开算子链。
 *          全局设置:env.disableOperatorChaining();
 *
 *          某个算子断开:
 *                  前后都断开:  算子.disableChaining()
 *                  和前面断开:  算子.startNewChain()
 *
 *
 */
public class Demo12_OperatorChain
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);
        //全局禁用算子链
        //env.disableOperatorChaining();

        DataStreamSource<String> ds = env.socketTextStream("hadoop103",8888);

        ds.map(line -> line).name("m1")
          .map(line -> line).name("m2").startNewChain()//.disableChaining()//.slotSharingGroup("a")
          .map(line -> line).name("m3")
          .print();

        env.execute();

    }
}
